// boolean condition = true;
// while(condition) {
//     // 客户端去拉取数据的时候，如果服务端没有数据响应，会保持连接等待服务器响应
//     // poll 中传入的超时时长参数，是指等待的最大时长
//     ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(long.MAX_VALUE));

//         // 对数据进行业务处理
//     // Iterator<ConsumerRecord<String, String>> iterator = records.iterarot();
//         // 这里可以获得迭代器，用 iter.啥next遍历
//     // 直接用 for 循环来迭代本次获取到的一批数据
//     for(ConsumerRecord<String, String> record : records) {
//         // ConsumerRecord 中，不光有用户的业务数据，还有 kafka 塞入的元数据
//         String key = record.key();
//         String value = record.value();

//         // 本条数据所属的 topic
//         String topic = record.topic();

//         // 所属分区
//         int partition = record.partition();

//         // offset 这条数据的偏移量
//         long offset = record.offset();

//         // 当前这条数据所在分区的 leader 的朝代几年
//         Optional<Integer> leaderEpoch = record.leaderEpoch();

//         // 在 kafka 的数据底层存储中，不光有用户的业务数据，还有大量元数据
//         // timestamp 就是其中之一，记录本条数据的时间戳
//         // 但是时间戳有两种类型：本条数据的创建时间（生产者）；本条数据的追加时间（broker写入log文件的时间）
//         TimestampType timestampType = record.timestampType();
//         long timestamp = record.timestamp();

//         // 数据头
//         // 数据头是生产者在写入数据时附加进去的（相当于用户自定义的元数据）
//         Headers headers = record.headers();

//         System.out.println(
//             String.format("数据key: %s, 数据value: %s, topic: %s, partition: %d, offset: %d, leader纪元: %s, timestampType: %s, timestamp",
//             key, value, topic, partition, offset, leaderEpoch.toString(), timestampType.name, timestamp)
//         );
//     }
// }

        // KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);    // 这里配置groupid 为doit30-3

        // consumer.subscribe(Arrays.asList("reb-1", "reb-2"), new ConsumerRebalanceListener() {   // 这里有个监听，就是回调方法
        //     // 再均衡过程中消费者会被取消先前所分配的主题, 分区
        //     // 取消了之后，consumer底层就会调用下面的方法
        //     public void onPartitionRevoked(Conllection<TopicPartition> partitions) {
        //         System.out.println("当前消费者被取消分区有: " + partitions);
        //     }

        //     // 再均衡过程中，消费者会被重新分配到新的主题, 分区
        //     // 分配好了新的主题，分区后，consumer底层调用下面的方法
        //     public void onPartitionAssigned(Conllection<TopicPartition> partitions) {
        //         System.out.println("当前消费者又被分配如下主题: " + partitions);
        //     }
        // });

        // while(true) {
        //     consumer.poll(Duration.ofMillis(long.MAX_VALUE));
        // }